package com.siyu.common.utils.grpc;

import java.util.concurrent.*;

public class QueueBasedAsyncProcessor {
    private final BlockingQueue<Runnable> queue;
    private final ThreadPoolExecutor executor;

    public QueueBasedAsyncProcessor(int threadCount, int queueSize) {
        queue = new LinkedBlockingQueue<>(queueSize);
        executor = new ThreadPoolExecutor(threadCount, threadCount,
                0L, TimeUnit.MILLISECONDS,
                queue,
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public void processAsync(Runnable task) {
        executor.execute(task);
    }

    public void shutdown() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}